package framework.transport.client;

import framework.registry.Registry;
import framework.registry.ServiceAddress;
import framework.transport.codec.XMessage;
import framework.transport.codec.XMessageCodec;
import framework.transport.codec.XRequest;
import framework.transport.codec.XResponse;
import framework.transport.loadbalance.LoadBalance;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

@Slf4j
public class NettyClient {


    private final Registry registry;

    private final LoadBalance loadBalance;

    private final ChannelProvider channelProvider;

    private final Bootstrap bootstrap;

    private final EventLoopGroup eventLoopGroup;

    public NettyClient(Registry registry, LoadBalance loadBalance, ChannelProvider provider) {
        this.registry = registry;
        this.loadBalance = loadBalance;
        this.channelProvider = provider;

        eventLoopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.DEBUG))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline()
                                .addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS))
                                .addLast(new LengthFieldBasedFrameDecoder(4 * 1024 * 1024, 6, 4, 0, 0))
                                .addLast(new XMessageCodec())
                                .addLast(new XClientHandler());
                    }
                });
    }


    public Channel getChannel(InetSocketAddress inetSocketAddress) throws Exception {

        Channel channel = channelProvider.get(inetSocketAddress);
        if (channel == null) {
            channel = doConnect(inetSocketAddress);
            channelProvider.set(inetSocketAddress, channel);
        }
        return channel;
    }

    public Channel doConnect(InetSocketAddress inetSocketAddress) throws Exception {
        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
        bootstrap.connect(inetSocketAddress)
                .addListener((ChannelFutureListener)future -> {
                    if (future.isSuccess()) {
                        log.debug("client connected [{}]", inetSocketAddress.toString());
                        completableFuture.complete(future.channel());
                    } else {
                        future.cause().printStackTrace();
                        throw new RuntimeException(String.format("client connect [%s] error", inetSocketAddress.toString()));
                    }
                });
        return completableFuture.get();
    }

    public XResponse<Object> sendRequest(XRequest request) throws Exception {

        List<ServiceAddress> serviceAddresses = registry.serviceDiscovery(request.getRpcServiceName());
        ServiceAddress select = loadBalance.select(serviceAddresses);
        InetSocketAddress inetSocketAddress = new InetSocketAddress(select.getHost(), select.getPort());
        Channel channel = getChannel(inetSocketAddress);
        if (!channel.isActive()) {
            throw new RuntimeException("channel not active");
        }
        // 构建异步返回值
        CompletableFuture<XResponse<Object>> resultFuture = new CompletableFuture<>();
        XMessage message = XMessage.getRequest(request);
        RequestPool.put(message.getRequestId(), resultFuture);
        channel.writeAndFlush(message).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.debug("client send message: [{}]", message);
            } else {
                future.channel().close();
                resultFuture.completeExceptionally(future.cause());
                log.error("Send failed:", future.cause());
            }
        });
        return resultFuture.get();
    }

}